package com.iflytek.controller;

import com.alibaba.fastjson.JSON;
import com.iflytek.bean.MonitorInfo;
import org.apache.avro.data.Json;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.shaded.guava18.com.google.common.cache.CacheBuilder;
import org.apache.flink.shaded.guava18.com.google.common.cache.CacheLoader;
import org.apache.flink.shaded.guava18.com.google.common.cache.LoadingCache;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import com.iflytek.utils.JdbcUtils;

public class _01OverSpeedController {
    public static void main(String[] args) throws Exception {
        //1.创建env对象
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 让 Flink 自动选择批处理或流处理模式。
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //2.消费kafka中的数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092");
        properties.setProperty("group.id", "group1");
        // 创建 FlinkKafkaConsumer 实例，从 topic-car 主题消费数据，使用 SimpleStringSchema 对数据进行序列化和反序列化。
        FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>("topic-car", new SimpleStringSchema(), properties);
        // 将 Kafka 消费者添加到 Flink 环境中，得到一个 DataStreamSource<String> 类型的数据流 ds1
        DataStreamSource<String> ds1 = env.addSource(consumer);
        // ds1.print();
        
        // 3. 将字符串转换成MonitorInfo对象
        // 使用 map 算子对 ds1 中的每个元素进行转换。
        SingleOutputStreamOperator<MonitorInfo> ds2 = ds1.map(
                // 定义一个匿名 MapFunction 实现类
                new MapFunction<String, MonitorInfo>() {
                    @Override
                    public MonitorInfo map(String s) throws Exception {
                        // 将 Kafka 接收到的 JSON 字符串使用 JSON.parseObject 方法转换为 MonitorInfo 对象。
                        return JSON.parseObject(s, MonitorInfo.class);
                    }
                }
        );
        // ds2.print();
        // 使用 filter 算子对 ds2 中的数据进行过滤。
        SingleOutputStreamOperator<MonitorInfo> ds3 =ds2.filter(
                // 定义一个匿名 RichFilterFunction 实现类，
                new RichFilterFunction<MonitorInfo>() {
                    Connection connection;
                    PreparedStatement ps;
                    ResultSet rs;
                    LoadingCache<String, Integer> cache;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        connection = JdbcUtils.getconnection();
                        // 根据卡口编号查询限速信息
                        ps = connection.prepareStatement("select speed_limit from t_monitor_info where monitor_id = ?");
                        // 使用 Guava 的 CacheBuilder 构建一个本地缓存对象
                        cache = CacheBuilder.newBuilder()
                                .maximumSize(100)                                   //缓存最大存储条数
                                .expireAfterWrite(100, TimeUnit.MINUTES)    //缓存过期时间
                                // 构建缓存对象，并传入一个 CacheLoader 实例，用于在缓存未命中时从数据库加载数据
                                .build(new CacheLoader<String, Integer>() {
                                    // load 方法在缓存中没有所需数据时被触发，用于从数据库中查询并加载数据到缓存
                                    // load 是在 别人问cache 要数据，假如缓存没有，就会触发 load 执行
                                    @Override
                                    public Integer load(String monitorId) throws Exception {
                                        System.out.println("查询数据库.......");
                                        // 为预编译 SQL 语句的占位符设置实际的卡口编号
                                        ps.setString(1, monitorId);
                                        // 执行 SQL 查询，并将结果集赋值给 rs 变量
                                        rs = ps.executeQuery();
                                        //如果t_monitor_info无法查询出该卡口的编号，则给定一个60的限速
                                        int speed_limit = 60;
                                        if (rs.next()) {
                                            speed_limit = rs.getInt("speed_limit");
                                        }
                                        // 返回查询到的限速信息，该信息会被缓存起来
                                        return speed_limit;
                                    }
                                });

                    }

                    @Override
                    public void close() throws Exception {
                        JdbcUtils.release(rs, ps, connection);
                    }

                    @Override
                    public boolean filter(MonitorInfo monitorInfo) throws Exception {
                        // 从缓存中获取当前卡口的限速信息。
                        // 使用 monitorInfo.getMonitorId() 作为键从缓存 cache 中获取对应的限速值。
                        // 如果缓存中存在该键对应的值，直接返回；若不存在，则会触发 CacheLoader 中的 load 方法从数据库查询并将结果存入缓存。
                        Integer speed_limit = cache.get(monitorInfo.getMonitorId());
                        // 将获取到的限速信息设置到 MonitorInfo 对象中，方便后续使用。
                        monitorInfo.setSpeedLimit(speed_limit);
                        // 若车辆的实际速度大于限速值的 1.1 倍（即超速 10%），则返回 true，表示该对象会被保留在数据流中；
                        // 否则返回 false，表示该对象会被过滤掉。
                        return monitorInfo.getSpeed() > speed_limit * 1.1;
                    }
                }
        );
        ds3.print();
        // 4. sink 数据输出
        ds3.addSink(JdbcSink.sink(
                "insert into t_speeding_info values(null,?,?,?,?,?,?)",
                (PreparedStatement ps, MonitorInfo monitorInfo) -> {
                    ps.setString(1, monitorInfo.getCar());
                    ps.setString(2, monitorInfo.getMonitorId());
                    ps.setString(3, monitorInfo.getRoadId());
                    ps.setDouble(4, monitorInfo.getSpeed());
                    ps.setInt(5, monitorInfo.getSpeedLimit());
                    ps.setLong(6, monitorInfo.getActionTime());
                },
                // 配置 JDBC 执行选项，withBatchSize(1) 表示批量写入的大小为 1，即每条数据都会立即写入；withBatchIntervalMs(5000) 表示批量写入的间隔为 5000 毫秒（5 秒）。
                JdbcExecutionOptions.builder().withBatchSize(1).withBatchIntervalMs(5000).build()
                ,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://hadoop102:3306/flink_project?useSSL=false&useUnicode=true&characterEncoding=utf8")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("12345678")
                        .build()));
        // 5.执行
        env.execute();
    }
}
